package cn.initcap.concurrency.concurrency.aqs;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;

/**
 * 适用于想要收集几个一起执行的场景
 *
 * @author initcap
 * @date Created in 2020/4/7 16:41.
 */
@Slf4j
public class CyclicBarrierExample {

    private static final int THREAD_SIZE = 10;

    /**
     * 创建线程池
     */
    private static final ExecutorService executorService = new ThreadPoolExecutor(0, THREAD_SIZE,
            60L, TimeUnit.SECONDS, new SynchronousQueue<>(),
            new ThreadFactoryBuilder().setNameFormat("CyclicBarrierExample").build());

    /**
     * 设置五个屏障的线程数,收集五个一起执行
     */
    private static CyclicBarrier barrier = new CyclicBarrier(5);
    /**
     * 设置五个屏障的线程数,收集五个一起执行,第二个参数是在等待中优先执行的runnable
     */
    private static CyclicBarrier barrierRunnable = new CyclicBarrier(5, () -> log.info("callbace is running"));

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < THREAD_SIZE; i++) {
            final int threadNum = i;
            Thread.sleep(1000);
            // executorService.execute(() -> race(threadNum));
            // executorService.execute(() -> raceAwait(threadNum));
            executorService.execute(() -> raceRunnable(threadNum));
        }
        executorService.shutdown();
    }

    private static void race(int threadNum) {
        try {
            Thread.sleep(1000);
            log.info("race : {} is ready", threadNum);
            barrier.await();
            log.info("race : {} continue", threadNum);
        } catch (InterruptedException e) {
            log.info("InterruptedException:{}", e);
            Thread.currentThread().interrupt();
        } catch (BrokenBarrierException e) {
            log.info("BrokenBarrierException:{}", e);
        }
    }

    private static void raceAwait(int threadNum) {
        try {
            Thread.sleep(1000);
            log.info("raceAwait : {} is ready", threadNum);
            barrier.await(2000, TimeUnit.MILLISECONDS);
            log.info("raceAwait : {} continue", threadNum);
        } catch (BrokenBarrierException | TimeoutException e) {
            log.warn("BarrierException", e);
        } catch (InterruptedException e) {
            log.info("InterruptedException", e);
            Thread.currentThread().interrupt();
        }
    }

    private static void raceRunnable(int threadNum) {
        try {
            Thread.sleep(1000);
            log.info("raceRunnable : {} is ready", threadNum);
            barrierRunnable.await();
            log.info("raceRunnable : {} continue", threadNum);
        } catch (InterruptedException e) {
            log.info("InterruptedException:{}", e);
            Thread.currentThread().interrupt();
        } catch (BrokenBarrierException e) {
            log.info("BrokenBarrierException:{}", e);
        }
    }

}
